package net.acoder.dz.transmit;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Queue;

public abstract class MyStream implements Runnable {
    protected List<Queue<byte[]>> outQueues    = new ArrayList<Queue<byte[]>>();
    protected Queue<byte[]>       inQueue;
    protected final int           bufferLenght = 256;
    protected String              name;
    protected boolean             running      = true;

    public abstract InputStream getInputStream();

    public abstract OutputStream getOutputStream();

    public void run() {
        System.out.println(name + " start");
        startRead();
        startWrite();
        System.out.println(name + " end");
    }

    public void startRead() {
        new Thread(new Runnable() {
            public void run() {
                byte[] buffer = new byte[bufferLenght];
                int len = -1;
                try {
                    while (running
                            && (len = getInputStream().read(buffer)) != -1) {
                        byte[] bs = new byte[len];
                        System.arraycopy(buffer, 0, bs, 0, len);
                        for (Queue<byte[]> queue : outQueues) {
                            queue.add(bs);
                            // System.out.println(Arrays.toString(bs));
                        }
                    }
                } catch (IOException e) {
                    System.out.println(name);
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void startWrite() {
        new Thread(new Runnable() {
            public void run() {
                try {
                    while (running) {
                        byte[] bs = inQueue.poll();
                        if (bs != null && bs.length > 0) {
                            getOutputStream().write(bs);
                            // System.out.println(Arrays.toString(bs));
                        } else {
                            // System.out.println(name +" bs is empty");
                            Thread.sleep(200);
                        }
                    }
                } catch (Exception e) {
                    System.out.println(name);
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void addOutQueue(Queue<byte[]> queue) {
        this.outQueues.add(queue);
    }

    public void setStoreQueue(Queue<byte[]> queue) {
        this.inQueue = queue;
    }

    public void stop() {
        running = false;
    }
}
